package com.gitee.xhs.keyedState.listState;

import com.gitee.xhs.EnvironmentConfig;
import com.gitee.xhs.FlinkEnvironmentFactory;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import com.gitee.xhs.sampleSource.TransactionBoundOneSecondSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static com.gitee.xhs.keyedState.listState.ListStateConstant.LIST_STATE_PATH;
import static com.gitee.xhs.keyedState.listState.ListStateConstant.LIST_STATE_UID;

/**
 * @author BigHead
 * @date 2022/6/30 10:09
 * @description ListStateMain
 */
public class ListStateMain {

    public static void main(String[] args) throws Exception {
        EnvironmentConfig environmentConfiguration = new EnvironmentConfig(null);
        environmentConfiguration.checkpoint(
                null,
                null,
                LIST_STATE_PATH,
                null
        );

        StreamExecutionEnvironment streamExecutionEnvironment = FlinkEnvironmentFactory.simpleEnv(environmentConfiguration);

        streamExecutionEnvironment
                .addSource(new TransactionBoundOneSecondSource())
                .keyBy(Transaction::getAccountId)
                .process(new ListStateKeyedProcessFunction()).uid(LIST_STATE_UID)
                .print();

        streamExecutionEnvironment.execute("ListStateMain");

    }

}
